package SparkGraphXInAction

import org.apache.spark.graphx._
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2017/4/25 0025.
  * 最短路径距离算法（包含最短路径）
  * 本代码记录了最短路径。
  */
object TestDijstraWithPath {
  //针对有向图的包含路径记录的Dijkstra最短路径算法
  def dijkstraTrace[VD](g: Graph[VD, Double], origin: VertexId)= {
    //1,初始化
    var g2 = g.mapVertices(
      (vid, vd) =>
        (false, if (vid == origin) 0 else Double.MaxValue,List[VertexId]())
    )

    //2，遍历所有的点
    for(i <- 1L to g.vertices.count -1){
      //3,确定最短路径值最小的作为当前顶点，当然了.filter(!_._2._1)操作，确保了筛选范围是没有访问过的顶点。
      val currentVertexId = g2.vertices.filter(!_._2._1).fold((0L,(false, Double.MaxValue,List[VertexId]()))){
        (a,b) => if (a._2._2 < b._2._2 ) a else b
      }._1

      //4，向与当前顶点相邻的顶点发消息，再聚合消息：取小值做为最短路径值；同时发送的还有路径轨迹。
      val newDistances: VertexRDD[(Double, List[VertexId])] =
        g2.aggregateMessages[(Double, List[VertexId])](
          //sendMsg函数，ctx.srcAttr._2是顶点的第二个值是距离，ctx.srcAttr._3是顶点的第三个值是存储的到origin的最短路径轨迹。
          ctx => if (ctx.srcId == currentVertexId) ctx.sendToDst((ctx.srcAttr._2 +ctx.attr, ctx.srcAttr._3 :+ ctx.srcId)),
          //mergeMsg函数,下面代码中a或者b是前一个阶段sendMsg函数发送的值。
          (a,b) => if (a._1 < b._1) a else b
        )

      //5,生成结果图。
      g2 = g2.outerJoinVertices(newDistances)((vid,vd,newSum) => {
        val newSumVal = newSum.getOrElse((Double.MaxValue, List[VertexId]()))
        (vd._1 || vid == currentVertexId, math.min(vd._2, newSumVal._1), if (vd._2 < newSumVal._1) vd._3 else newSumVal._2)
      })
    }


    g.outerJoinVertices(g2.vertices){ (vid, vd, dist) =>
      (vd, dist.getOrElse((false, Double.MaxValue,List[VertexId]()))) //这里之前报了类型错误，这是由于我前面把函数的返回值类型规定错了，删掉返回值类型就好了
    }
  }

  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val myVertices = sc.makeRDD(Array((1L,"A"),(2L,"B"),(3L,"C"),(4L,"D"),(5L,"E"),(6L,"F"),(7L,"G")))
    val myEdges = sc.makeRDD(Array(Edge(1L,2L,7.0),Edge(1L,4L,5.0),Edge(2L,3L,8.0),Edge(2L,4L,9.0),
      Edge(2L,5L,7.0),Edge(3L,5L,5.0),Edge(4L,5L,15.0),Edge(4L,6L,6.0),
      Edge(5L,6L,8.0),Edge(5L,7L,9.0),Edge(6L,7L,11.0)))
    val myGraph = Graph(myVertices,myEdges)
//    //展示图
//    println("展示图：")
//    val list_triplets = myGraph.triplets.collect
//    for(kk <- list_triplets){
//      println(kk)
//    }

    //执行dijkstraTrace
    val list_path_distance = dijkstraTrace(myGraph,1L).vertices.map(_._2).collect
    for(x <- list_path_distance){
      println(x)
    }
  }
}
